package com.gzc.just.play.last.war.sceneserver.network.broadcast;

import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 广播队列 - 用于存储待发送的广播消息
 * 
 * 映射自C++ 消息队列系统
 * 采用优先级队列，支持并发访问
 * 
 * @author AI编程助手
 * @version 1.0
 */
public class BroadcastQueue {
    
    private static class PrioritizedMessage implements Comparable<PrioritizedMessage> {
        BroadcastMessage message;
        long sequenceNumber;
        
        PrioritizedMessage(BroadcastMessage message, long sequenceNumber) {
            this.message = message;
            this.sequenceNumber = sequenceNumber;
        }
        
        @Override
        public int compareTo(PrioritizedMessage other) {
            // 优先级高的优先处理
            int priorityCompare = Integer.compare(other.message.getPriority(), this.message.getPriority());
            if (priorityCompare != 0) {
                return priorityCompare;
            }
            // 优先级相同时，按序列号排序
            return Long.compare(this.sequenceNumber, other.sequenceNumber);
        }
    }
    
    private final PriorityBlockingQueue<PrioritizedMessage> queue;
    private final AtomicLong sequenceNumber = new AtomicLong(0);
    private final AtomicLong totalEnqueued = new AtomicLong(0);
    private final AtomicLong totalDequeued = new AtomicLong(0);
    private final int maxQueueSize;
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    
    /**
     * 构造函数
     * 
     * @param maxQueueSize 队列最大容量
     */
    public BroadcastQueue(int maxQueueSize) {
        this.maxQueueSize = maxQueueSize;
        this.queue = new PriorityBlockingQueue<>(maxQueueSize);
    }
    
    /**
     * 添加消息到队列
     * 
     * @param message 广播消息
     * @return 是否添加成功
     */
    public boolean enqueue(BroadcastMessage message) {
        if (message == null) {
            return false;
        }
        
        lock.readLock().lock();
        try {
            if (queue.size() >= maxQueueSize) {
                return false; // 队列满
            }
            
            PrioritizedMessage prioritized = new PrioritizedMessage(
                    message,
                    sequenceNumber.incrementAndGet()
            );
            
            boolean success = queue.offer(prioritized);
            if (success) {
                totalEnqueued.incrementAndGet();
            }
            return success;
        } finally {
            lock.readLock().unlock();
        }
    }
    
    /**
     * 从队列获取消息（非阻塞）
     * 
     * @return 广播消息，如果队列为空则返回null
     */
    public BroadcastMessage dequeue() {
        lock.readLock().lock();
        try {
            PrioritizedMessage prioritized = queue.poll();
            if (prioritized != null) {
                totalDequeued.incrementAndGet();
                return prioritized.message;
            }
            return null;
        } finally {
            lock.readLock().unlock();
        }
    }
    
    /**
     * 从队列获取消息（阻塞）
     * 
     * @return 广播消息
     * @throws InterruptedException 如果线程被中断
     */
    public BroadcastMessage dequeueBlocking() throws InterruptedException {
        lock.readLock().lock();
        try {
            PrioritizedMessage prioritized = queue.take();
            if (prioritized != null) {
                totalDequeued.incrementAndGet();
                return prioritized.message;
            }
            return null;
        } finally {
            lock.readLock().unlock();
        }
    }
    
    /**
     * 查看队列中的下一条消息（不移除）
     * 
     * @return 下一条消息，如果队列为空则返回null
     */
    public BroadcastMessage peek() {
        lock.readLock().lock();
        try {
            PrioritizedMessage prioritized = queue.peek();
            return prioritized != null ? prioritized.message : null;
        } finally {
            lock.readLock().unlock();
        }
    }
    
    /**
     * 获取队列大小
     * 
     * @return 当前队列中的消息数
     */
    public int size() {
        lock.readLock().lock();
        try {
            return queue.size();
        } finally {
            lock.readLock().unlock();
        }
    }
    
    /**
     * 队列是否为空
     * 
     * @return 是否为空
     */
    public boolean isEmpty() {
        return size() == 0;
    }
    
    /**
     * 清空队列
     */
    public void clear() {
        lock.writeLock().lock();
        try {
            queue.clear();
        } finally {
            lock.writeLock().unlock();
        }
    }
    
    /**
     * 获取最大队列容量
     * 
     * @return 最大容量
     */
    public int getMaxQueueSize() {
        return maxQueueSize;
    }
    
    /**
     * 获取总入队消息数
     * 
     * @return 入队总数
     */
    public long getTotalEnqueued() {
        return totalEnqueued.get();
    }
    
    /**
     * 获取总出队消息数
     * 
     * @return 出队总数
     */
    public long getTotalDequeued() {
        return totalDequeued.get();
    }
    
    /**
     * 获取队列统计信息
     * 
     * @return 统计字符串
     */
    public String getStatistics() {
        lock.readLock().lock();
        try {
            return String.format("BroadcastQueue{size=%d, maxSize=%d, totalEnqueued=%d, totalDequeued=%d}",
                               queue.size(), maxQueueSize, totalEnqueued.get(), totalDequeued.get());
        } finally {
            lock.readLock().unlock();
        }
    }
}
